package com.tsx.kafka.product;
import java.util.Properties;
import java.util.concurrent.Future;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
public class KafkaUtil {
 
    private static KafkaProducer<String, String> producer=null;  
   
    /**
     * 采用单例模式 
     */
    private static synchronized void getInstallKafka()  {
    	 if(producer==null){
    		 synchronized (KafkaUtil.class) {
    			 Properties props = new Properties();
    			 /**
    			  *  （用于建立与kafka集群连接的host/port组。数据将会在所有servers上均衡加载，不管哪些server是指定用于bootstrapping。这个列表仅仅影响初始化的hosts（用于发现全部的servers）。这个列表格式：
    			  *	host1:port1,host2:port2,...
    			  *	因为这些server仅仅是用于初始化的连接，以发现集群所有成员关系（可能会动态的变化），这个列表不需要包含所有的servers（你可能想要不止一 个server，尽管这样，可能某个server宕机了）。如果没有server在这个列表出现，则发送数据会一直失败，直到列表可用。）
    			  */
//    			 props.put("bootstrap.servers", "172.17.6.10:9092,172.17.6.11:9092,172.17.6.12:9092");// kafka
//    			 props.put("zookeeper.connect", "172.17.6.10:2181,172.17.6.11:2181,172.17.6.12:2181");//声明zk 
    			  props.put("bootstrap.servers", "172.16.6.206:9092");// kafka 单机版
    			  props.put("zookeeper.connect", "172.16.6.206:2181");//zookeeper单机版
    			 /**
    			  * （producer需要server接收到数据之后发出的确认接收的信号，此项配置就是指procuder需要多少个这样的确认信号。此配置实际上代表了数据备份的可用性。以下设置为常用选项：
    			  *（1）acks=0： 设置为0表示producer不需要等待任何确认收到的信息。副本将立即加到socket  buffer并认为已经发送。没有任何保障可以保证此种情况下server已经成功接收数据，同时重试配置不会发生作用（因为客户端不知道是否失败）回 馈的offset会总是设置为-1；
    			  *（2）acks=1： 这意味着至少要等待leader已经成功将数据写入本地log，但是并没有等待所有follower是否成功写入。这种情况下，如果follower没有成功备份数据，而此时leader又挂掉，则消息会丢失。
    			  *（3）acks=all： 这意味着leader需要等待所有备份都成功写入日志，这种策略会保证只要有一个备份存活就不会丢失数据。这是最强的保证。
    			  *（4）其他的设置，例如acks=2也是可以的，这将需要给定的acks数量，但是这种策略一般很少用。）
    			  */
    			 props.put("acks", "all");
    			 props.put("group.id", "xinyun_study");//必须要使用别的组名称， 如果生产者和消费者都在同一组，则不能访问同一组内的topic数据  
    			 props.put("retries", 0);
    			 props.put("batch.size", 16384);
    			 props.put("linger.ms", 1);
    			 props.put("buffer.memory", 33554432);
    			 //   props.put("compression.type", "none");// 用于压缩数据的压缩类型。默认是无压缩。正确的选项值是none、gzip、snappy。 压缩最好用于批量处理，批量处理消息越多，压缩性能越好。
    			 // props.put("client.id", "172.17.6.13"); 标明数据入库请求来源
    			 props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    			 props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); 
    			 producer = new KafkaProducer<>(props);	
    		 }
    	 }
    }
    
    static{  
        if(producer==null){
        	System.out.println("---kafka 进入创建 product实例中----");
        	getInstallKafka();
        }
    }  
  
   

    /**
     *  
     * @param topic 存储数据的主题
     * @param partition  
     * @param key  存储数据的分组的key
     * @param value 存储数据内容
     * @return
     * @throws Exception
     */
    public static Future<RecordMetadata> send(String topic, Integer partition, String key, String value) throws Exception{  
        try {  
        	 if(producer==null)
        	 getInstallKafka();
        	 return   producer.send(new ProducerRecord<String, String>(topic,partition, key, value));
        }catch(Exception e){  
            throw e;  
        }  
    }  
    
    public static  void colseProduct(){
    	producer.close();
    	System.err.println("--kafka进入销毁 product 进程完成-----");
    }
   
}
